package com.shujia.flink.window;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo03CountWindow {
    public static void main(String[] args) throws Exception {
        /* 3、基于计数的窗口：可以直接在KeyBy之后直接.出来
         *       滑动：
         *       滚动：
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<String> wordDS = env.socketTextStream("master", 8888);

        wordDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(kv -> kv.f0)
//                .countWindow(5) // 每个key每5条数据统计一次
                .countWindow(10, 5) // 每个key每5条数据统计最近的10条数据
                .sum(1)
                .print();

        env.execute();

    }
}
